<?php

use Swoole\Coroutine;
use Swoole\Coroutine\Redis;

// 配置参数
$redisHost = '192.168.100.27';
$redisPort = 6379;
$streamKey = 'xnh5_consume_pay_callback_queue';
$groupName = 'xnh5_consume_group';
$consumerName = 'xnh5_consume_group_1_consumer_worker_test';
$maxRetries = 5; // 最大重试次数
$minIdleTime = 1000; // 30秒
$blockTimeout = 5000; // 5秒

/**
 * 初始化 Redis 连接
 */
function initRedis($host, $port) {
    $redis = new \Swoole\Coroutine\Redis();
    $redis->connect($host, $port);
    $redis->select(2);
    return $redis;
}

/**
 * 消费者逻辑：处理单条消息
 */
function processMessage($redis, $stream, $group, $id, $data) {
    try {
        // 模拟业务逻辑
        echo "正在处理消息 ID: {$id}\n";

        // 模拟可能失败的情况
        if (rand(0, 1)) {
            throw new \Exception("模拟处理失败");
        }

        // 成功处理后确认
        $redis->xAck($stream, $group, [$id]);
        echo "已确认消息 ID: {$id}\n";
        return true;
    } catch (\Exception $e) {
        error_log("处理消息失败: " . $e->getMessage());
        return false;
    }
}

/**
 * 定时任务：自动认领超时消息
 */
function autoClaimPendingMessages($redis, $stream, $group, $consumer, $minIdleTime, $maxRetries) {
    while (true) {
        // 自动认领超时消息
        $claimed = $redis->xAutoClaim(
            $stream,
            $group,
            $consumer,
            $minIdleTime,
            '0-0',
            ['COUNT' => 10]
        );

        foreach ($claimed['messages'] as $msg) {
            $id = $msg[0];
            $data = $msg[1];

            // 获取重试次数
            $retryCount = isset($data['retries']) ? intval($data['retries']) : 0;

            if ($retryCount >= $maxRetries) {
                // 达到最大重试次数，告警
                error_log("消息 ID: {$id} 已达到最大重试次数 {$maxRetries}");
                continue;
            }

            // 更新重试次数
            $data['retries'] = $retryCount + 1;

            // 处理消息
            if (processMessage($redis, $stream, $group, $id, $data)) {
                echo "成功处理并确认消息 ID: {$id}\n";
            } else {
                echo "处理失败，将重试消息 ID: {$id}\n";
            }
        }

        // 每隔一段时间检查一次
        Coroutine::sleep(30);
    }
}

/**
 * 启动消费者并处理消息
 */
function startConsumer($redis, $stream, $group, $consumer, $maxRetries, $blockTimeout) {
    while (true) {
        // 读取消息
        $messages = $redis->xReadGroup(
            $group,
            $consumer,
            [$stream => '>'],
            ['count' => 1, 'block' => $blockTimeout]
        );

        if ($messages === false) {
            $error = $redis->getLastError();
            error_log("读取消息失败: $error");
            continue;
        }

        foreach ($messages[$stream] as $msg) {
            $id = $msg[0];
            $data = $msg[1];

            // 获取重试次数
            $retryCount = isset($data['retries']) ? intval($data['retries']) : 0;

            if ($retryCount >= $maxRetries) {
                // 达到最大重试次数，告警
                error_log("消息 ID: {$id} 已达到最大重试次数 {$maxRetries}");
                continue;
            }

            // 处理消息
            if (processMessage($redis, $stream, $group, $id, $data)) {
                echo "成功处理并确认消息 ID: {$id}\n";
            } else {
                echo "处理失败，将重试消息 ID: {$id}\n";
            }
        }
    }
}

// 初始化 Redis 连接


// 启动两个协程：一个处理新消息，一个处理卡住的消息
go(function () use ($redisHost,$redisPort,$minIdleTime, $streamKey, $groupName, $consumerName, $maxRetries, $blockTimeout) {
    //startConsumer($redis, $streamKey, $groupName, $consumerName, $maxRetries, $blockTimeout);
    $redis = initRedis($redisHost, $redisPort);
    $stream = 'xnh5_consume_pay_callback_queue';
    $group = 'xnh5_consume_group';
    $rescueConsumer = 'rescue-worker-' . uniqid();
    $minIdleTime = 1000; // 30秒

// 认领超过 30 秒未处理的消息
    /*$claimed = $redis->xClaim(
        $stream,
        $group,
        $rescueConsumer,
        $minIdleTime,
        ['1758270648871-0']
    );
    print_r($claimed);*/
    $claimed = $redis->xAutoClaim(
        $stream,
        $group,
        'xnh5_consume_group_1_consumer_worker0ddd',
        $minIdleTime,
        '0'
    );
    print_r($claimed);
    //echo "成功认领 {$claimed['next_id']} 条消息\n";
    var_dump($streamKey);
    var_dump($groupName);
    var_dump($consumerName);
    var_dump($claimed);
});

go(function () use ($streamKey, $groupName, $consumerName, $minIdleTime, $maxRetries) {
    //autoClaimPendingMessages($redis, $streamKey, $groupName, $consumerName, $minIdleTime, $maxRetries);
});